package com.wang.event;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;

public class OrderedConsumer<T> implements Consumer<Event<T>> {
    private BlockingQueue<Event<T>> list = new LinkedBlockingQueue<>();
    private volatile boolean isRunning = false;
    private Executor executor;
    private Consumer<Event<T>> consumer;


    public OrderedConsumer(Executor executor, Consumer<Event<T>> consumer) {
        this.consumer = consumer;
        this.executor = executor;
    }

    @Override
    public void accept(Event<T> event) {
        list.add(event);
        if(this.isRunning) {
            return ;
        }
        synchronized (this) {
            if(!this.isRunning) {
                this.isRunning = true;
                this.run();
            }
        }
    }

    public void run() {
        this.executor.execute(()->{
            Event<T> event;
            while ((event = list.peek()) != null) {
                consumer.accept(event);
            }
            synchronized (this) {
                boolean remain = list.size() == 0;
                if(remain) {
                    this.run();
                } else {
                    this.isRunning = false;
                }
            }
        });
    }
}